package com.pie4cloud.pie.bus.service.listener;

import com.alibaba.nacos.shaded.com.google.gson.Gson;
import com.pie4cloud.pie.bus.api.consumer.factory.KfkClientFactory;
import com.pie4cloud.pie.bus.api.dto.ConsumerDto;
import com.pie4cloud.pie.bus.api.feign.RemoteKafkaConsumerService;
import com.pie4cloud.pie.bus.config.FeignRemote;
import com.pie4cloud.pie.common.core.constant.SecurityConstants;
import com.pie4cloud.pie.common.core.util.R;
import feign.Request;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;

import java.time.Duration;
import java.util.*;
import java.util.concurrent.*;

@Component
@Slf4j
public class KafkaConsumerListener {
	@Autowired
	private FeignRemote feignRemote;
	@Value(value = "${spring.kafka.consumer.auto-commit-interval}")
	private int autoCommitInterval;
	@Value(value = "${spring.kafka.bootstrap-servers}")
	private String bootstrap;
	@Value(value = "${spring.kafka.consumer.auto-offset-reset}")
	private String autoOffsetReset;
	@Value(value = "${spring.kafka.consumer.enable-auto-commit}")
	private Boolean enableAutoCommit;
	@Value(value = "${spring.kafka.consumer.key-deserializer}")
	private String keyDeserializer;
	@Value(value = "${spring.kafka.consumer.value-deserializer}")
	private String valueDeserializer;
	private volatile boolean  isRunning = true;


	private ExecutorService cachedThreadPool = Executors.newCachedThreadPool();
	private ExecutorService fixedThreadPool = Executors.newFixedThreadPool(1);
	//private BlockingQueue<ConsumerDto> queue=new LinkedBlockingQueue();


	@KafkaListener(topics = "bus", groupId = "bus",containerFactory = "batchFactoryEarliest")
	public void topicBus(ConsumerRecord<?, ?> record, Acknowledgment ack, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) {
		Optional message = Optional.ofNullable(record.value());
		if (message.isPresent()) {
			String msg = message.get().toString();
			ConsumerDto consumerDto=new Gson().fromJson(msg, ConsumerDto.class);
			try {
				//queue.put(consumerDto);
				cachedThreadPool.execute(
						()->{
							this.doPoll(consumerDto);
						});
				ack.acknowledge();
			} catch (Exception e) {
				e.printStackTrace();
			}
			log.info("topic_test 消费了： Topic:" + topic + ",Message:" + msg);


		/*	max.poll.records=9999
			max.partition.fetch.bytes=104857600
			fetch.min.bytes=2097152
			fetch.max.wait.ms=10000
			fetch.max.bytes=104857600*/
			//props.put("fetch.min.bytes", "2097152");


			//1.创建消费者
			//KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);

			//2.订阅Topic
			//创建一个只包含单个元素的列表，Topic的名字叫作customerCountries
			//consumer.subscribe(Collections.singletonList(consumerDto.getTopic()));
			//支持正则表达式，订阅所有与test相关的Topic
			//consumer.subscribe("test.*");

			//3.轮询
			//消息轮询是消费者的核心API，通过一个简单的轮询向服务器请求数据，一旦消费者订阅了Topic，轮询就会处理所欲的细节，包括群组协调、partition再均衡、发送心跳
			//以及获取数据，开发者只要处理从partition返回的数据即可。

			//ConsumerThread consumerThread=new ConsumerThread(consumer,remoteKfkConsumerService,consumerDto,props);
			//consumerThread.setName(consumerDto.getTopic());
			//consumerThread.start();

		}
	}

	@KafkaListener(topics = "stop", groupId = "bus"+"-" + "#{T(java.util.UUID).randomUUID()})",containerFactory = "batchFactoryLatest")
	public void topicStop(ConsumerRecord<?, ?> record, Acknowledgment ack, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) {
		Optional message = Optional.ofNullable(record.value());
		if (message.isPresent()) {
			String msg = message.get().toString();
			ConsumerDto consumerDto=new Gson().fromJson(msg, ConsumerDto.class);
			String key=consumerDto.getKafGroup()+"_"+consumerDto.getTopic();
			Long value=Long.parseLong(consumerDto.getTime());
			KfkClientFactory.isStop.put(key,value);
			ack.acknowledge();
		}
	}
   /* public void pollQueue(){
		fixedThreadPool.execute(()->{
			while (isRunning){
				try {
					ConsumerDto consumerDto=queue.take();
					cachedThreadPool.execute(
							()->{
								this.doPoll(consumerDto);
							}
							//new ConsumerThread(remoteKfkConsumerService,consumerDto,props)
					);
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			}
		});

	}*/
	public void doPoll (ConsumerDto consumerDto){
		Properties props = new Properties();
		props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrap);
		props.put("group.id", consumerDto.getKafGroup());
		props.put("key.deserializer", keyDeserializer);
		props.put("value.deserializer", valueDeserializer);
		//props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG,autoCommitInterval);
		props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,false);
		props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"earliest");
		props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG,100000);
		props.put(ConsumerConfig.REQUEST_TIMEOUT_MS_CONFIG,110000);
		props.put("max.poll.records",1000);
		props.put("fetch.max.wait.ms", 10000);
		props.put("fetch.max.bytes", 104857600);
		KafkaConsumer<String, String> consumer = new KafkaConsumer(props);
		consumer.subscribe(Collections.singletonList(consumerDto.getTopic()));
		String key=consumerDto.getKafGroup()+"_"+consumerDto.getTopic();
		Long value = Long.parseLong(consumerDto.getTime());
		KfkClientFactory.isStop.put(key, value);
		RemoteKafkaConsumerService remoteKafkaConsumerService =feignRemote.target(RemoteKafkaConsumerService.class,consumerDto.getServiceName());
		while (true){
			if (value < KfkClientFactory.isStop.get(key)) {
				consumer.unsubscribe();
				consumer.close();
				break;
			}
			ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(10000));
			Map<TopicPartition, OffsetAndMetadata> currentOffsets = new HashMap();
			Map<TopicPartition, Long> offsets = new HashMap();

			boolean flag=true;
			for (ConsumerRecord<String, String> record : records) {
				try {
					consumerDto.setContent(record.value());
					Request.Options options = new Request.Options(3, TimeUnit.HOURS, 3, TimeUnit.HOURS, true);

					R r = remoteKafkaConsumerService.consume(options, consumerDto, SecurityConstants.FROM_IN);
					offsets.put(new TopicPartition(record.topic(),record.partition()),record.offset()+1);
					currentOffsets.put(new TopicPartition(record.topic(), record.partition()),
							new OffsetAndMetadata(record.offset() + 1, "no metadata"));

				} catch (Exception e) {
					flag=false;
					break;
				}

			}
			//consumer.commitSync();
			if(currentOffsets.size()>0){
				consumer.commitSync(currentOffsets);
				currentOffsets.clear();
			}
			if(!flag){
				records.forEach(record -> {
					offsets.computeIfAbsent(new TopicPartition(record.topic(),record.partition()),keyTp->record.offset());
				});
				for (Map.Entry<TopicPartition, Long> entry : offsets.entrySet()) {
					consumer.seek(entry.getKey(),entry.getValue());
				}
				offsets.clear();
				try {
					Thread.sleep(10000);
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			}
			Thread.yield();

		}
	}
}
